import com.teradata.tap.metadata.Column;
import com.teradata.tap.metadata.SchemaElement;
import com.teradata.tap.metadata.settings.Setting;
import com.teradata.trm.pe.common.task.ProcessingEngineTask;
import com.teradata.trm.pe.common.util.ProcessingEngineUtils;
import com.teradata.trm.common.db.Result;
import com.teradata.trm.common.util.Strings;
import com.teradata.trm.pe.common.RunDetails;
import com.teradata.trm.common.email.*;
import com.teradata.trm.common.context.ContextUtil;

import com.teradata.trm.common.workflow.*;
import com.teradata.trm.securitymanager.domain.User;
import com.teradata.trm.securitymanager.util.SecurityManagerUtil;
import java.util.Enumeration;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;
import java.util.List;
import java.text.DecimalFormat;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.io.*;


public class Waterfall extends ProcessingEngineTask {
    
    private Properties properties;

    private String singleBreak = "\r\n";
    private String doubleBreak = "\r\n\r\n";
    private String newRow = "<tr>";
    private String endRow = "</tr>";
    private int major = 4;
    private int minor = 0;

    private String certified = "Developed and certified for TRM version 6.2.0.x";

    private String SCR_EMAIL_TEMPLATE                   = "waterfall-email-template.html";
    private String SQL_COUNT_COMM_DEDUP                 = "waterfall-get-dedup-count.sql";
    private String SQL_COUNT_COMM_INVALID_ADDRESS       = "waterfall-get-invalid-address-count.sql";
    private String SQL_COUNT_CHANNEL_LIMIT              = "waterfall-get-channel-count.sql";
    private String SQL_COUNT_COMM_TOTAL_LIMIT           = "waterfall-get-comm-total_count.sql";
    private String SQL_COUNT_COMM_CYCLE_LIMIT           = "waterfall-get-comm-cycle-count.sql";
    private String SQL_COUNT_RECENCY_TOO_SOON_LIMITS    = "waterfall-get-recency-too-soon-count.sql";
    private String SQL_COUNT_RECENCY_TOO_MANY_LIMITS    = "waterfall-get-recency-too-many-count.sql";
    private String SQL_COMM_DETAILS                     = "waterfall-get-comm-details.sql";
    private String[] segments = null;
    private String[] tables = null;
    private String[] operationalSegments = null;
    private String[] operationalTables = null;
    private String[] operationalOperators = null;

    private boolean[] dedupes = null;
    private int planNet = 0;
    private int SNP,SN0, SN1, SN2, SN3, SN4 = 0;
    private int DO00, DO01, DO1, DO2, DO3, DO4, D05, DO3DO4 = 0;
    private long start = System.currentTimeMillis();
    private String recencyHtml = "";

    @Override
    public Serializable execute() throws TaskException {
        try {
            String dash = repeat('-', 81);
            String tab = repeat(' ', 10);
            String headline = doubleBreak + dash
                + doubleBreak + tab + "TRM Waterfall Extension v" + major + "." + minor
                + doubleBreak + dash + singleBreak;
            String footer = doubleBreak + dash
                + doubleBreak + tab + "TRM Waterfall Extension"
                + doubleBreak + tab + "Process Complete"
                + doubleBreak + dash + singleBreak;

            RunDetails details = this.getRunDetails();
            properties = new Properties();
            String runId = details.getRunId();

            info(headline);

            addProperty("RunId", runId);

            List<String> objectIds = ProcessingEngineUtils.getActiveObjectIds(details.getJobHistory());

            User user = SecurityManagerUtil.getUser(details.getJobSubmitUser());
            addProperty("User", user.getUserName());
            addProperty("FullName", user.getFirstName() + " " + user.getLastName());
            addProperty("EmailToAddress", user.getEmail());
            addProperty("CommId",objectIds.get(0));
            String log = doubleBreak + "User requesting waterfall count:" +
                    doubleBreak + "Full name     = " + getProperty("FullName") +
                    singleBreak + "User name     = " + getProperty("User") +
                    singleBreak + "Email Address = " + getProperty("EmailToAddress") + singleBreak;
            info(log);
            addProperty("Application", "Waterfall Extension version " + String.valueOf(major) + "." + String.valueOf(minor));
            getSegmentPlanDetails(getProperty("CommId"));
            runWaterfall();
            info(footer);
            return 0;
        }
        catch (Exception ex) {
            String msg = doubleBreak + repeat('*', 81) + doubleBreak;
            msg += "WATERFALL HAS FAILED" + doubleBreak;
            msg += "Reason:" + singleBreak + ex.getMessage() + doubleBreak;
            msg += "Stack trace:" + singleBreak;
            for (int i = 0; i < ex.getStackTrace().length; i++)
                msg += ex.getStackTrace()[i].toString() + singleBreak;
            msg += doubleBreak + repeat('*', 81) + doubleBreak;
            info(msg);
            sendEmailFailure(ex);
            return 0;
        }
        finally {
            shutdown();
        }
    }

    private void runWaterfall() throws Exception {
        segments = null;
        tables = null;
        dedupes = null;

        if ((getRunDetails().getProcessingEngineId().equals("SSPE00000002")) && getPlanDetails()){
            gatherAllSegments();
            prepareCounter();
            prepareDedupe();
            getSegments();

            for (int i = 0; i < segments.length; i++)
            {
                addProperty("LogicalSegmentId", segments[i]);
                addProperty("SegmentTable", tables[i]);
                addProperty("Dedupe", String.valueOf(dedupes[i]));
                addProperty("TempSegment", "Wc_Temp_" + segments[i]);

                prepareSegment("@SpindleTemp", "@TempSegment");

                operationalSegments = null;
                operationalTables = null;
                operationalOperators = null;

                getOperationalSegments();

                if (operationalSegments != null) {
                    for (int j = 0; j < operationalSegments.length; j++) {
                        addProperty("OperationalSegmentId", operationalSegments[j]);
                        addProperty("OperationalSegmentTable", operationalTables[j]);
                        addProperty("Operator", operationalOperators[j]);
                        addProperty("Order", String.valueOf(j+1));
                        operationalCount();
                    }
                }
                finalizeSegment(i);
                dropTable("@SpindleTemp", "@TempSegment");
            }

            planNet = getCount("select count(*) from @SpindleTemp.@DedupeList");
            applyCommDedup();
            applyInvalidAddress();
            applyRecencyLimiting();
            applyChannelLimiting();
            applyCommLimiting();

            sendReport();

            dropTable("@SpindleTemp", "@CountList");
            dropTable("@SpindleTemp", "@DedupeList");
            dropTable("@SpindleTemp", "@SegmentList");
            dropTable("@SpindleTemp", "@OperationalList");
            dropTable("@SpindleTemp", "@RecencyCounts");
            dropTable("@SpindleTemp", "@RecencyContacts");
        }
        else
            info(doubleBreak + "Waterfall counts not enabled for segment plan" + singleBreak);

    }
    private int getCount(String sql) throws TaskException {
        sql = replaceProperties(sql);
        int count = 0;
        Result result = runQuery(sql);
        if (result.hasErrors())
            throw new TaskException("Failed to get count: " + result.getErrors().get(0));
        ResultSet set = result.getResultSet();
        try {
            if (set != null) {
                while (set.next()) {
                    count = set.getInt(1);
                }
            }
        }
        catch (Exception e) {
            throw new TaskException(e.getMessage());
        }
        result.discard();
        return count;
    }

    private void applyCommDedup() throws TaskException {
        String body = loadScript(SQL_COUNT_COMM_DEDUP);
        DO00 = getCount(body);
        info("Total deduplicated leads:" + formatNumber(DO1));
        SNP = planNet - DO00;
    }

    private void applyInvalidAddress() throws TaskException {
        String body = loadScript(SQL_COUNT_COMM_INVALID_ADDRESS);
        DO01 = getCount(body);
        info("Total invalid address leads:" + formatNumber(DO1));
        SN0 = SNP - DO01;
        DO1 = DO00 + DO01;
    }

    private void applyCommLimiting() throws TaskException {
        String body = loadScript(SQL_COUNT_COMM_CYCLE_LIMIT);
        DO3 = getCount(body);
        info("Total communication cycle limited leads:" + formatNumber(DO3));
        SN3 = SN2 - DO3;
        body = loadScript(SQL_COUNT_COMM_TOTAL_LIMIT);
        DO4 = getCount(body);
        info("Total communication total max limited leads:" + formatNumber(DO4));
        DO3DO4 = DO3+DO4;
        SN4 = SN3 - DO4;
    }

    private void applyChannelLimiting() throws TaskException {
        String body = loadScript(SQL_COUNT_CHANNEL_LIMIT);
        DO2 = getCount(body);
        info("Total channel limited leads:" + formatNumber(DO2));
        SN2 = SN1 - DO2;

    }

    private void prepareCounter() throws TaskException {
        String sql = "Create Table @SpindleTemp.@CountList (Logical_Segment_Id Varchar(12), Operational_Segment_Id Varchar(12), Display_Ord Integer, Reduction_Count Integer, Total_Count Integer) Primary Index (Logical_Segment_Id)";
        dropTable("@SpindleTemp", "@CountList");
        executeUpdateSql(sql);
    }

    private void prepareDedupe() throws TaskException {
        String sql = "Create Table @SpindleTemp.@DedupeList As (Select @Groupby From @Spindletemp.Lh_Current_Lead Where (1 = 0)) With No Data Primary Index (@Groupby)";
        dropTable("@SpindleTemp", "@DedupeList");
        executeUpdateSql(sql);
    }

    private void gatherAllSegments() throws TaskException {

        dropTable("@SpindleTemp", "@SegmentList");
        dropTable("@SpindleTemp", "@OperationalList");

        String sql = "Create Table @SpindleTemp.@SegmentList As (Select T1.Logical_Segment_Id, T2.Segment_Id, T2.Name, T1.Display_Ord, T3.Database_Name_Txt, T3.Table_Name_Txt, T1.Deduplicate_Ind, T1.Deduplicate_Ord From @Metadata.Sm_Logical_Segment T1 Join @Metadata.Sm_Segment T2 On T1.Segment_Id = T2.Segment_Id Join @SpindleTemp.Rt_Segment_Header T3 On T1.Logical_Segment_Id = T3.Logical_Segment_Id Where T1.Segment_Plan_Id = '@PlanId' And T3.Run_Id = '@RunId' And T1.Segment_Role_Cd = 0) With Data Primary Index (Logical_Segment_Id)";
        executeUpdateSql(sql);

        sql = "Create Table @SpindleTemp.@OperationalList As (Select T2.Logical_Segment_Id, T2.Segment_Id, T2.Segment_Role_Cd, T4.Name, T2.Display_Ord, T3.Database_Name_Txt, T3.Table_Name_Txt, T1.Child_Relationship_Id From @Metadata.Sm_Segment_Relationship T1 Join @Metadata.Sm_Logical_Segment T2 On T1.Parent_Relationship_Id = T2.Logical_Segment_Id And T1.Segment_Plan_Id = T2.Segment_Plan_Id Join @SpindleTemp.Rt_Segment_Header T3 On T2.Logical_Segment_Id = T3.Logical_Segment_Id Join @Metadata.Sm_Segment T4 On T2.Segment_Id = T4.Segment_Id Where T1.Segment_Plan_Id = '@PlanId' And T3.Run_Id = '@RunId') With Data Primary Index(Logical_Segment_Id)";
        executeUpdateSql(sql);
    }

    private void getSegments() throws TaskException {
        String sql = "Select Logical_Segment_Id, Database_Name_Txt, Table_Name_Txt, Deduplicate_Ind From @SpindleTemp.@SegmentList Order By Deduplicate_Ord";
        sql = replaceProperties(sql);
        Result result = runQuery(sql);
        if (result.hasErrors())
            throw new TaskException("Failed to get segment details: " + result.getErrors().get(0));

        ResultSet set = result.getResultSet();
        try {
            while (set.next()) {
                if (segments == null) {
                    segments = new String[1];
                    tables = new String[1];
                    dedupes = new boolean[1];
                    segments[0] = set.getString(1);
                    tables[0] = set.getString(2).trim() + "." + set.getString(3).trim();
                    dedupes[0] = (set.getInt(4) == 1);
                }
                else {
                    String[] t1 = new String[segments.length + 1];
                    String[] t2 = new String[segments.length + 1];
                    boolean[] t3 = new boolean[segments.length + 1];
                    for (int i = 0; i < segments.length; i++) {
                        t1[i] = segments[i];
                        t2[i] = tables[i];
                        t3[i] = dedupes[i];
                    }
                    t1[segments.length] = set.getString(1);
                    t2[segments.length] = set.getString(2).trim() + "." + set.getString(3).trim();
                    t3[segments.length] = (set.getInt(4) == 1);
                    segments = t1;
                    tables = t2;
                    dedupes = t3;
                }
            }

        } catch (SQLException sex) {
            throw new TaskException(sex.getMessage());
        }
        if (set != null)
            result.discard();
    }

    private void getOperationalSegments() throws TaskException {
        String sql = "Select Logical_Segment_Id, Database_Name_Txt, Table_Name_Txt, Segment_Role_Cd From @SpindleTemp.@OperationalList Where Child_Relationship_Id = '@LogicalSegmentId' Order By Display_Ord";
        sql = replaceProperties(sql);
        Result result = runQuery(sql);
        if (result.hasErrors())
            throw new TaskException("Failed to get operational segment details: " + result.getErrors().get(0));

        ResultSet set = result.getResultSet();
        try {
            while (set.next()) {
                if (operationalSegments == null) {
                    operationalSegments = new String[1];
                    operationalTables = new String[1];
                    operationalOperators = new String[1];
                    operationalSegments[0] = set.getString(1);
                    operationalTables[0] = set.getString(2).trim() + "." + set.getString(3).trim();
                    operationalOperators[0] = (set.getInt(4) == 5 ? "Not In" : "In");
                }
                else {
                    String[] t1 = new String[operationalSegments.length + 1];
                    String[] t2 = new String[operationalSegments.length + 1];
                    String[] t3 = new String[operationalSegments.length + 1];
                    for (int i = 0; i < operationalSegments.length; i++) {
                        t1[i] = operationalSegments[i];
                        t2[i] = operationalTables[i];
                        t3[i] = operationalOperators[i];
                    }
                    t1[operationalSegments.length] = set.getString(1);
                    t2[operationalSegments.length] = set.getString(2).trim() + "." + set.getString(3).trim();
                    t3[operationalSegments.length] = (set.getInt(4) == 5 ? "Not In" : "In");
                    operationalSegments = t1;
                    operationalTables = t2;
                    operationalOperators = t3;
                }
            }

        } catch (SQLException sex) {
            throw new TaskException(sex.getMessage());
        }
        if (set != null)
            result.discard();
    }

    private boolean getPlanDetails() throws TaskException {
        boolean doWaterfall = false;
        String sql = loadScript(SQL_COMM_DETAILS);
        sql = replaceProperties(sql);
        Result result = runQuery(sql);
        if (result.hasErrors())
            throw new TaskException("Failed to get segment plan details: " + result.getErrors().get(0));
        if (result.hasRows()) {
            ResultSet set = result.getResultSet();
            try {
                while (set.next()) {
                    int level = set.getInt("Schema_Element_Id");
                    if (level > 0) {
                        SchemaElement se = getMetadataService().getSchemaElement(level);

                        addProperty("CustomerLevel", se.getName());
                        for(Column column : se.getMasterTableKeyColumns()) {
                            addProperty("Groupby", column.getName());
                        }
                        addProperty("CommName"             ,set.getString("CommName"));
                        addProperty("DeduplicationTypeCd"  ,set.getString("DeduplicationTypeCd"));
                        addProperty("DeduplicationFrom"    ,set.getString("DeduplicationFrom"));
                        addProperty("DedupType"            ,set.getString("DedupType"));
                        addProperty("RecencyClassName"     ,set.getString("RecencyClassName"));
                        addProperty("RecencyClassId"       ,set.getString("RecencyClassId"));
                        addProperty("RecencySubClassId"    ,set.getString("RecencySubClassId"));
                        addProperty("CycleLimit"           ,formatNumber(set.getInt("CycleLimit")));
                        addProperty("CommMax"              ,formatNumber(set.getInt("CommMax")));
                        addProperty("ChannelName"          ,set.getString("ChannelName"));
                        addProperty("ControlPct"           ,set.getString("ControlPct"));
                        doWaterfall = true;
                    }
                }
            } catch (SQLException sex) {
                throw new TaskException(sex.getMessage());
            }
        }
        if (result != null)
            result.discard();
        return doWaterfall;
    }

    private void prepareSegment(String database, String table) throws TaskException {
        dropTable(database, table);
        String sql1 = "Create Table " + database + "." + table + " As (Select @Groupby From @SegmentTable Group By @Groupby) With Data Unique Primary Index (@Groupby)";
        String sql2 = "Insert Into @SpindleTemp.@CountList (Logical_Segment_Id, Display_Ord, Reduction_Count, Total_Count) Select '@LogicalSegmentId', 0, 0, Count(*) From @SpindleTemp.@TempSegment";
        executeUpdateSql(sql1);
        executeUpdateSql(sql2);
    }

    private void operationalCount() throws TaskException {
        String sql = "";
        sql = "Delete From @SpindleTemp.@TempSegment Where @Groupby @Operator (Select @Groupby From @OperationalSegmentTable)";
        executeUpdateSql(sql);
        sql = "Insert Into @SpindleTemp.@CountList (Logical_Segment_Id, Operational_Segment_Id, Display_Ord, Reduction_Count, Total_Count) Select '@LogicalSegmentId', '@OperationalSegmentId', @Order, 0, Count(*) From @SpindleTemp.@TempSegment";
        executeUpdateSql(sql);
        sql = "Update T1 From @SpindleTemp.@CountList T1, (Select Total_Count From @SpindleTemp.@CountList Where Logical_Segment_Id = '@LogicalSegmentId' And Display_Ord = @Order - 1) T2 Set Reduction_Count = T2.Total_Count - T1.Total_Count Where T1.Logical_Segment_Id = '@LogicalSegmentId' And T1.Display_Ord = @Order";
        executeUpdateSql(sql);
    }

    private void finalizeSegment(int index) throws TaskException {
        String sql = "";

        if (dedupes[index]) {
            sql = "Delete From @SpindleTemp.@TempSegment Where @Groupby In (Select @Groupby From @SpindleTemp.@DedupeList)";
            executeUpdateSql(sql);
            sql = "Insert Into @SpindleTemp.@CountList (Logical_Segment_Id, Display_Ord, Reduction_Count, Total_Count) Select '@LogicalSegmentId', 998, 0, Count(*) From @SpindleTemp.@TempSegment";
            executeUpdateSql(sql);
            sql = "Update T1 From @SpindleTemp.@CountList T1, (Select Max(Display_Ord) Last_Ord From @SpindleTemp.@CountList Where Logical_Segment_Id = '@LogicalSegmentId' And Display_Ord < 998) T2 , (Select Display_Ord, Total_Count From @SpindleTemp.@CountList Where Logical_Segment_Id = '@LogicalSegmentId' And Display_Ord < 998 Group By 1, 2) T3 Set Reduction_Count = T3.Total_Count - T1.Total_Count Where T1.Logical_Segment_Id = '@LogicalSegmentId' And T1.Display_Ord = 998 And T3.Display_Ord = T2.Last_Ord";
            executeUpdateSql(sql);
        }
        sql = "Insert Into @SpindleTemp.@DedupeList (@Groupby) Select @Groupby From @SpindleTemp.@TempSegment";
        executeUpdateSql(sql);

        sql = "Insert Into @SpindleTemp.@CountList ( Logical_Segment_Id, Display_Ord, Reduction_Count, Total_Count) Select '@LogicalSegmentId', 999, 0, Count(*) From @SpindleTemp.@TempSegment";
        executeUpdateSql(sql);

        sql = "Update T1 From @SpindleTemp.@CountList T1, (Select Sum(Reduction_Count) Reduction_Count From @SpindleTemp.@CountList Where Logical_Segment_Id = '@LogicalSegmentId' And Display_Ord Between 1 And 998) T2 Set Reduction_Count = T2.Reduction_Count Where T1.Logical_Segment_Id = '@LogicalSegmentId' And T1.Display_Ord = 999";
        executeUpdateSql(sql);

    }

    private void applyRecencyLimiting() throws TaskException {
        String html = "";
        SN1=SN0;
        String sql = loadScript(SQL_COUNT_RECENCY_TOO_SOON_LIMITS);
        sql = replaceProperties(sql);
        Result result = runQuery(sql);
        if (result.hasErrors())
            throw new TaskException("Failed to get recency results report data: " + result.getErrors().get(0));
        ResultSet set = result.getResultSet();
        html += "<table border=\"0\" class=\"report\">";
        html += newRow;
        html += addCell("title"         ,"left",    "Contact Optimization");
        html += addCell("titleName"     ,"left",    "Class");
        html += addCell("titleName"     ,"left",    "Subclass");
        html += addCell("titleDropoff"  ,"right",   "Drop Off");
        html += addCell("titleNet"      ,"right",   "Net");
        html += endRow;
        html += newRow;
        html += addCell("normalCell", "left", "After communication deduplication");
        html += addCell("normalCell", "left", "");
        html += addCell("normalCell", "left", "");
        html += addCell("normalCell", "right", "0");
        html += addCell("normalCell", "right", formatNumber(SN1));
        html += endRow;
        try {
            while (set.next()) {
                html += newRow;
                html += addCell("normalCell", "left", "");
                html += addCell("normalCell", "left", emptyIfNull(set.getString("RecencyClass")));
                html += addCell("normalCell", "left", emptyIfNull(set.getString("RecencySubclass")));
                html += addCell("normalCell", "right", formatNumber(set.getInt("RecencyDropOff")));
                SN1 = SN1 - set.getInt("RecencyDropOff");
                html += addCell("normalCell", "right", formatNumber(SN1));
                html += endRow;
            }
        } catch (SQLException sex) {
                throw new TaskException(sex.getMessage());
        }
        if (set != null)
        result.discard();
        if (SN0 > SN1) {
            html += newRow;
            html += addCell("normalCell", "left", "");
            html += addCell("normalCell", "left", "");
            html += addCell("normalCell", "left", "");
            html += addCell("totalCell", "right", formatNumber(SN0 - SN1));
            html += addCell("totalCell", "right", formatNumber(SN1));
            html += "</table>";
        } else {
            html = "";
        }
        recencyHtml = html;
    }
    private void sendReport() throws TaskException {

        Emailer emailService = null;

        String sql = "Select T1.Logical_Segment_Id, T2.Name Segment_Name, T2.Segment_Id, T3.Segment_Id Operational_Segment_Id, T3.Name Operational_Name, Cast(Case When T3.Segment_Role_Cd = 5 Then 'Universe' When T3.Segment_Role_Cd = 6 Then 'Suppression' When T3.Segment_Role_Cd = 7 Then 'Control' When T3.Segment_Role_Cd Is Null And T1.Display_Ord = 0 Then 'Segment' Else Null End As Varchar(20)) Operational_Role, T1.Reduction_Count, T1.Total_Count, T1.Display_Ord Segment_Ord, T2.Display_Ord From @SpindleTemp.@CountList T1 Join @SpindleTemp.@SegmentList T2 On T1.Logical_Segment_Id = T2.Logical_Segment_Id Left Outer Join @SpindleTemp.@OperationalList T3 On T1.Logical_Segment_Id = T3.Child_Relationship_Id And T1.Operational_Segment_Id = T3.Logical_Segment_Id Order By T2.Display_Ord, T1.Display_Ord";
        sql = replaceProperties(sql);
        Result result = runQuery(sql);
        if (result.hasErrors())
            throw new TaskException("Failed to get report data: " + result.getErrors().get(0));

        ResultSet set = result.getResultSet();
        String html = "";
        addProperty("SegmentationHtml", html);
        int lastSegment = -1;
        String dataset = "";
        dataset = "Segment_Order,Segment_Name,Segment_Id,Operational_Segment_Name,Operational_Segment_Id,Segment_Role,Segment_Order,Dropoff_Count,Total_Count" + singleBreak;
        try {
            while (set.next()) {
                if (set.getInt("DISPLAY_ORD") != lastSegment) {
                    if (lastSegment >= 0)
                       html += "</table><p>";
                    html += "<table border=\"0\" class=\"report\">";
                    html += newRow;
                    html += addCell("titleName", "left", "Segment");
                    html += addCell("titleId", "left", "Id");
                    html += addCell("titleRole", "left", "Role");
                    html += addCell("titleDropoff", "right", "Drop Off");
                    html += addCell("titleNet", "right", "Net");
                    html += endRow;
                    lastSegment = set.getInt("DISPLAY_ORD");
                }

                int segmentOrd = set.getInt("SEGMENT_ORD");
                html += newRow;
                if (segmentOrd == 0) {
                    html += addCell("normalCell", "left", emptyIfNull(set.getString("SEGMENT_NAME")));
                    html += addCell("normalCell", "left", emptyIfNull(set.getString("SEGMENT_ID")));
                    html += addCell("normalCell", "left", "Segment");
                    html += addCell("normalCell", "right", formatNumber(set.getInt("REDUCTION_COUNT")));
                    html += addCell("normalCell", "right", formatNumber(set.getInt("TOTAL_COUNT")));
                }

                else if (segmentOrd < 998) {
                    html += addCell("normalCell", "left", emptyIfNull(set.getString("OPERATIONAL_NAME")));
                    html += addCell("normalCell", "left", emptyIfNull(set.getString("OPERATIONAL_SEGMENT_ID")));
                    html += addCell("normalCell", "left", emptyIfNull(set.getString("OPERATIONAL_ROLE")));
                    html += addCell("normalCell", "right", formatNumber(set.getInt("REDUCTION_COUNT")));
                    html += addCell("normalCell", "right", formatNumber(set.getInt("TOTAL_COUNT")));
                }
                else if (segmentOrd == 998) {
                    html += addCell("normalCell", "left", "&nbsp;");
                    html += addCell("normalCell", "left", "&nbsp;");
                    html += addCell("normalCell", "left", "Deduplication");
                    html += addCell("normalCell", "right", formatNumber(set.getInt("REDUCTION_COUNT")));
                    html += addCell("normalCell", "right", formatNumber(set.getInt("TOTAL_COUNT")));
                }
                else {
                    html += addCell("normalCell", "left", "&nbsp;");
                    html += addCell("normalCell", "left", "&nbsp;");
                    html += addCell("normalCell", "left", "&nbsp;");
                    html += addCell("totalCell", "right", formatNumber(set.getInt("REDUCTION_COUNT")));
                    html += addCell("totalCell", "right", formatNumber(set.getInt("TOTAL_COUNT")));

                }


                html += endRow;
                dataset += "\"" + set.getInt("DISPLAY_ORD") + "\",";
                dataset += "\"" + emptyIfNull(set.getString("SEGMENT_NAME")) + "\",";
                dataset += "\"" + emptyIfNull(set.getString("SEGMENT_ID")) + "\",";
                dataset += "\"" + emptyIfNull(set.getString("OPERATIONAL_NAME")) + "\",";
                dataset += "\"" + emptyIfNull(set.getString("OPERATIONAL_SEGMENT_ID")) + "\",";
                dataset += "\"" + emptyIfNull(set.getString("OPERATIONAL_ROLE")) + "\",";
                dataset += "\"" + set.getInt("SEGMENT_ORD") + "\",";
                dataset += "\"" + set.getInt("REDUCTION_COUNT") + "\",";
                dataset += "\"" + set.getInt("TOTAL_COUNT") + "\"" + singleBreak;

            }
            if (html.length() > 1)
                html += "</table>";
        } catch (SQLException sex) {
            throw new TaskException(sex.getMessage());
        }
        if (set != null)
            result.discard();

        addProperty("PlanNet"   , formatNumber(planNet));
        addProperty("DO00"      , formatNumber(DO00));
        addProperty("DO01"      , formatNumber(DO01));
        addProperty("DO1"       , formatNumber(DO1));
        addProperty("DO2"       , formatNumber(DO2));
        addProperty("DO3"       , formatNumber(DO3));
        addProperty("DO4"       , formatNumber(DO4));
        addProperty("DO3DO4"    , formatNumber(DO3DO4));
        addProperty("SNP"       , formatNumber(SNP));
        addProperty("SN0"       , formatNumber(SN0));
        addProperty("SN1"       , formatNumber(SN1));
        addProperty("SN2"       , formatNumber(SN2));
        addProperty("SN3"       , formatNumber(SN3));
        addProperty("SN4"       , formatNumber(SN4));
        addProperty("SegmentationHtml", html);
        addProperty("RecencyHtml", recencyHtml);

        EmailImpl email = new EmailImpl();
        Setting setting = getMetadataService().getSetting("SMTP_FROM_TXT");
        if (setting != null)
            email.setFromEmail(setting.getSettingValue());
        else
            email.setFromEmail("trm@freemail.com");
        email.setToEmail(getProperty("EmailToAddress"));
	String subject = "Simulation results for " + getProperty("CommName") + " communication, id " + getProperty("CommId");
        addProperty("CurrentTime", getTimestamp());
        addProperty("ElapsedTime", getElapsedTimeHoursMinutesSecondsString(start));

        String body = loadScript(SCR_EMAIL_TEMPLATE);

	body = replaceProperties(body);
        email.setText(body);
        email.setSubject(subject);

        //try {
        //    String fileName = "waterfall-counts-" + getProperty("PlanId") + ".csv";
        //    if ((new File(fileName)).exists())
        //        (new File(fileName)).delete();
        //    FileWriter file = new FileWriter(fileName, true);
        //    PrintWriter printer = new PrintWriter(file);
        //    printer.print(dataset);
        //    printer.close();
        //    email.addAttachment(fileName);
        //}
        //catch (Exception e) {
        //    info("Failed to add attachment to email: " + e.getMessage());
        //}
        info("Email:" + singleBreak + body);
        try {emailService = ContextUtil.getBean("emailService");} catch(Throwable t){};
        if (emailService != null) {
            info("Email service is enabled. Sending email");
            emailService.sendEmail(email);
        }

    }

    private String addCell(String style, String align, String value) {
        String html = "<td ";
        html += "align=\"" + align + "\" ";
        html += "valign=\"top\" class=\"" + style + "\">";
        html += value + "</td>";
        return html;
    }

    private void addProperty(String name, String value) throws TaskException {
        try {
            properties.put(name, value);
        }
        catch (Exception ex) {
            throw new TaskException("Failed to add property (" + name + ", " + value + "): " + ex.getMessage());
        }
    }

    private String emptyIfNull(String string) {
        String result = "";
        if (string != null)
            result = string;
        return result;
    }

    private String formatNumber(int value) {
        DecimalFormat fmt = new DecimalFormat("###,###,##0");
        return fmt.format(value);
    }

    private String getProperty(String name) {
        String value = "";
        try {
            value = (String)properties.getProperty(name);
            if (value == null)
                value = "'" + name + "' was not found";
        }
        catch (Exception e) {}
        return value;
    }

    private String replaceProperties(String script) {
        String replaced = script;
        Enumeration lst = properties.propertyNames();
        for (; lst.hasMoreElements(); ) {
            String propName = (String)lst.nextElement();
            String propValue = (String)properties.get(propName);
            replaced = Strings.replace(replaced, "@" + propName, propValue);
        }
        return replaced;
    }

    private static String repeat(char c, int i) {
        String tst = "";
        for(int j = 0; j < i; j++) tst = tst+c;
        return tst;
    }

    private void info(String value) {
        getLogger().info(value);
    }

    private void display(String script) {
        String dash = repeat('-', 81);
        info(doubleBreak + dash + singleBreak + " Executing " + script + singleBreak + dash + singleBreak);
    }

    private void runCustomScript(String script) throws TaskException {
        display(script);
        String sql = this.loadScript(script);
        sql = replaceProperties(sql);
        int result = this.runScript(sql);
        if (result != 0)
            throw new TaskException(singleBreak + "Waterfall script '" + script + "' contained a failure");
    }

    private String getTimestamp() {
        String time = "";
        DateFormat fmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        time = fmt.format(new Date());
        return time;
    }
    private String getElapsedTimeHoursMinutesSecondsString(long startTime) {
        long elapsedTime = System.currentTimeMillis() - startTime;
        String format = String.format("%%0%dd", 2);
        elapsedTime = elapsedTime / 1000;
        String seconds = String.format(format, elapsedTime % 60);
        String minutes = String.format(format, (elapsedTime % 3600) / 60);
        String hours = String.format(format, elapsedTime / 3600);
        String time =  hours + ":" + minutes + ":" + seconds;
        return time;
    }
    private void dropTable(String database, String table) throws TaskException {
        String sql1 = "Select 'exists' Table_Exists From Dbc.Tables Where DatabaseName = '" + database + "' And TableName = '" + table + "' ";
        String sql2 = "Drop Table " + database + "." + table;
        sql1 = replaceProperties(sql1);
        sql2 = replaceProperties(sql2);
        Result result1 = runQuery(sql1);
        if (result1.hasErrors())
            throw new TaskException("Failed to read dbc dictionary: " + result1.getErrors().get(0));
        ResultSet set = result1.getResultSet();
        try {
            if (set != null) {
                while (set.next()) {
                    Result result2 = runQuery(sql2);
                    if (result2.hasErrors())
                        throw new TaskException("Failed to drop table: " + result2.getErrors().get(0));
                    result2.discard();
                }
            }
        }
        catch (Exception e) {
            throw new TaskException(e.getMessage());
        }
        result1.discard();
    }


    private void executeUpdateSql(String sql) throws TaskException {
        String sql1 = replaceProperties(sql);
        Result result1 = runQuery(sql1);
        if (result1.hasErrors())
            throw new TaskException("Failed to run sql: " + result1.getErrors().get(0));
        result1.discard();
    }

    private void getSegmentPlanDetails(String commId){
        String sql = "select segment_plan_id, name from @Metadata.sm_segment_plan where segment_plan_id in ";
        sql +=       " (select segment_plan_id from @Metadata.cm_communication where communication_id = '" + commId + "')";
        sql = replaceProperties(sql);
        Result result = runQuery(sql);
        String planId, planName;
        if (result.hasErrors())
            throw new TaskException("Failed to get segment plan details: " + result.getErrors().get(0));
        ResultSet set = result.getResultSet();
        try {
            while (set.next()) {
                planId =   set.getString(1);
                planName = set.getString(2);
                addProperty("PlanId", planId);
                addProperty("PlanName", planName);
                addProperty("SegmentList", "Wc_Segment_" + planId);
                addProperty("OperationalList", "Wc_Operational_" + planId);
                addProperty("CountList", "Wc_Count_" + planId);
                addProperty("DedupeList", "Wc_Dedupe_" + planId);
                info("Segment plan id   = " + set.getString(1));
                info("Segment plan name = " + set.getString(2));

            }
        } catch (SQLException sex) {
            throw new TaskException(sex.getMessage());
        }
        if (set != null)
            result.discard();
    }
    private void sendEmailFailure(Exception e) {
        Emailer emailService = null;
        try {
            String body = "<html><style>body {font-family: verdana; font-size: 11px}</style><body>";
            body += "Communication simulation failed for communication &quot;" + getProperty("PlanName") + "&quot; (id = " + getProperty("PlanId") + ")<p>";
            body += "<u>Reason</u><br>" + e.getMessage() + "<p>";
            body += "<div style=\"font-size: 9px; font-family: arial\">" + getProperty("Application") + "</div></body></html>";
            String fileName = "stacktrace.txt";
            if ((new File(fileName)).exists())
                (new File(fileName)).delete();
            FileWriter file = new FileWriter(fileName, true);
            PrintWriter printer = new PrintWriter(file);
            for (int i = 0; i < e.getStackTrace().length; i++)
                printer.println(e.getStackTrace()[i].toString());
            printer.close();
            EmailImpl email = new EmailImpl();
            Setting setting = getMetadataService().getSetting("SMTP_FROM_TXT");
            if (setting != null)
                email.setFromEmail(setting.getSettingValue());
            else
                email.setFromEmail("trm@freemail.com");
            email.setToEmail(getProperty("EmailToAddress"));
            String subject = "Waterfall counts failed for segment plan " + getProperty("PlanId");
            body = replaceProperties(body);
            email.setText(body);
            email.setSubject(subject);
            email.addAttachment(fileName);
            try {emailService = ContextUtil.getBean("emailService");} catch(Throwable t){};
            if (emailService != null) {
                info("Email service is enabled. Sending email");
                emailService.sendEmail(email);
            }
        }
        catch (Exception sex) {
            // just ignore the error
        }
    }
}
